package flinkcdc

import java.util.Properties

import com.ververica.cdc.connectors.mysql.source.MySqlSource
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.streaming.api.scala._

object P1_StreamCdc {

  def main(args: Array[String]): Unit = {

    val debeziumProperties = new Properties()
    debeziumProperties.setProperty("","")
    val mySqlSource = MySqlSource.builder[String]
      .hostname("bigdata001")
      .port(3306)
      .databaseList("flink")
      .tableList("flink.flink_cdc")
      .username("root")
      .password("root123U")
      .deserializer(new JsonDebeziumDeserializationSchema)
      .debeziumProperties(debeziumProperties)
      .startupOptions(StartupOptions.initial())
      .build // converts SourceRecord to JSON String


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.enableCheckpointing(5000)


    val mysqlCdcSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source").setParallelism(1)  // use parallelism 1 for sink to keep message ordering
    mysqlCdcSource.print()

    env.execute("cdc test")

  }

}
